#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import json
import time
from kafka import KafkaProducer
from datetime import datetime

def create_test_event(data_value, event_id):
    """创建测试事件"""
    return {
        "id": f"test-{event_id}",
        "type": "TEST",
        "source": "test-source",
        "timestamp": int(time.time() * 1000),
        "data": str(data_value),
        "severity": "INFO",
        "userId": "test-user"
    }

def send_test_data():
    """发送测试数据到Kafka"""
    try:
        print("正在连接Kafka服务器 localhost:9092...")
        # 创建Kafka生产者
        producer = KafkaProducer(
            bootstrap_servers=['localhost:9092'],
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            request_timeout_ms=10000,
            api_version=(0, 10, 1),
            retries=3
        )
        print("Kafka连接成功!")
        
        print("开始发送测试数据...")
        
        # 发送一些非1的数据
        for i in range(3):
            event = create_test_event(0, f"warm-up-{i}")
            producer.send('test', value=event)
            print(f"发送事件: {event}")
            time.sleep(1)
        
        # 发送连续3个1
        print("\n发送连续3个1...")
        for i in range(3):
            event = create_test_event(1, f"ones-{i}")
            producer.send('test', value=event)
            print(f"发送事件: {event}")
            time.sleep(1)
        
        # 发送一些其他数据
        for i in range(2):
            event = create_test_event(2, f"after-{i}")
            producer.send('test', value=event)
            print(f"发送事件: {event}")
            time.sleep(1)
        
        producer.flush()
        producer.close()
        print("\n测试数据发送完成!")
        
    except Exception as e:
        print(f"发送数据时出错: {e}")
        print(f"错误类型: {type(e).__name__}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    send_test_data()